package com.gopay.scheduler.job;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import javax.annotation.Resource;

import org.hibernate.criterion.DetachedCriteria;
import org.hibernate.criterion.Restrictions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.gopay.common.constants.cps.TrialBalanceConstants;
import com.gopay.common.cps.dao.acct.AcpsAcctTransDtlTmpDAO;
import com.gopay.common.cps.dao.acct.AcpsCalcBalanceDAO;
import com.gopay.common.cps.dao.acct.AcpsCalcBalanceErrDtlDAO;
import com.gopay.common.cps.dao.acct.CommonQueryDAO;
import com.gopay.common.cps.manager.trans.AcctTransManager;
import com.gopay.common.domain.acps.AcpsAcctTransDtl;
import com.gopay.common.domain.acps.AcpsAcctTransDtlTmp;
import com.gopay.common.domain.acps.AcpsCalcBalance;
import com.gopay.common.domain.acps.AcpsCalcBalanceErrDtl;
import com.gopay.common.domain.acps.Drcr;
import com.gopay.common.exception.GopayException;
import com.gopay.common.util.DateUtils;
import com.gopay.scheduler.GopayBaseJob;
import com.gopay.scheduler.job.vo.TrialBalanceVo;

/**
 * 试算平衡
 * 
 * @DailyTrialBalanceJob.java
 * @author lizheng
 * @2014-3-13 下午2:15:56 www.gopay.com.cn Inc.All rights reserved.
 */
public class DailyTrialBalanceJob extends GopayBaseJob {
    @Resource(name = "acctTransManager")
    private AcctTransManager acctTransManager;
    @Resource(name = "acpsAcctTransDtlTmpDAO")
    private AcpsAcctTransDtlTmpDAO acpsAcctTransDtlTmpDAO;
    @Resource(name = "acpsCalcBalanceDAO")
    private AcpsCalcBalanceDAO acpsCalcBalanceDAO;
    @Resource(name = "acpsCalcBalanceErrDtlDAO")
    private AcpsCalcBalanceErrDtlDAO acpsCalcBalanceErrDtlDAO;
    @Resource(name = "comQueryDao")
    private CommonQueryDAO commonQueryDAO;
    private Logger logger = LoggerFactory.getLogger(DailyTrialBalanceJob.class);

    private BlockingQueue<TrialBalanceVo> singleTranQueue;
    private AtomicLong pendingVisits = new AtomicLong();
    private long calcBalanceId;
    private String drSchema;
    private String crSchema;
    private int threadNum;
    private int queueSize;

    public void doJobInternal() {
        logger.info("daily trial balance start");
        Date currhDate = new Date();
        String yestedayDate = DateUtils.date2string(DateUtils.add(currhDate, Calendar.DATE, -1), "yyyy-MM-dd");
        Date startDate = DateUtils.strToDate(yestedayDate + " 00:00:00", "yyyy-MM-dd HH:mm:ss");
        Date endDate = DateUtils.strToDate(yestedayDate + " 23:59:59", "yyyy-MM-dd HH:mm:ss");

        trialProcess(startDate, endDate);
        logger.info("daily trial balance end");
    }

    /**
     * 试算主流程
     * 
     * @param startDate
     * @param endDate
     */
    public void trialProcess(Date startDate, Date endDate) {
        try {
            check(startDate, endDate);
        } catch (GopayException e) {
            logger.error(e.getErrMsg());
            e.printStackTrace();
            return;
        }

        AcpsCalcBalance bal = createAcpsCalcBalance(startDate, endDate);
        acpsCalcBalanceDAO.save(bal);

        calcBalanceId = bal.getId();

        // 期初余额
        TrialBalanceVo openBalanceVo = openBalance(startDate, endDate);
        bal.setBeginDr(openBalanceVo.getDrAmt());
        bal.setBeginCr(openBalanceVo.getCrAmt());
        bal.setModified(new Date());
        acpsCalcBalanceDAO.update(bal);

        // 期末余额
        TrialBalanceVo closingBalanceVo = closingBalance();
        bal.setEndDr(closingBalanceVo.getDrAmt());
        bal.setEndCr(closingBalanceVo.getCrAmt());
        bal.setModified(new Date());
        acpsCalcBalanceDAO.update(bal);

        // 初始化帐务明细数据
        initializeTmp(startDate, endDate);

        // 本期发生额
        TrialBalanceVo tranCalculateVo = periodAmt();
        bal.setCurrD(tranCalculateVo.getDrAmt());
        bal.setCurrC(tranCalculateVo.getCrAmt());
        bal.setIsSuccess(TrialBalanceConstants.CALC_BALANCE_SUCCESS);
        bal.setStatus(TrialBalanceConstants.CALC_BALANCE_STATUS_FIN);
        bal.setModified(new Date());
        acpsCalcBalanceDAO.update(bal);
    }

    private void check(Date startDate, Date endDate) throws GopayException {
        if (startDate == null || endDate == null)
            throw new GopayException("", "startDate or endDate can't null");
        if (startDate.after(endDate))
            throw new GopayException("", "startDate can't after endDate");
        if (startDate.after(new Date()))
            throw new GopayException("", "startDate can't after now");
        if (endDate.after(new Date()))
            throw new GopayException("", "endDate can't after now");

        DetachedCriteria detachedCriteria = DetachedCriteria.forClass(AcpsCalcBalance.class);
        detachedCriteria.add(Restrictions.eq("beginDate", startDate));
        detachedCriteria.add(Restrictions.eq("endDate", endDate));
        int count = acpsCalcBalanceDAO.findCountByCriteria(detachedCriteria);
        if (count > 0) {
            throw new GopayException("", "task has been deal");
        }
    }

    private AcpsCalcBalance createAcpsCalcBalance(Date startDate, Date endDate) {
        AcpsCalcBalance bal = new AcpsCalcBalance();
        bal.setId(acpsCalcBalanceDAO.getNextSq());
        bal.setBeginCr(new BigDecimal(0.0));
        bal.setBeginDr(new BigDecimal(0.0));
        bal.setCreated(new Date());
        bal.setCurrC(new BigDecimal(0.0));
        bal.setCurrD(new BigDecimal(0.0));
        bal.setEndCr(new BigDecimal(0.0));
        bal.setEndDr(new BigDecimal(0.0));
        bal.setIsSuccess(TrialBalanceConstants.CALC_BALANCE_FAIL);
        bal.setModified(new Date());
        bal.setBeginDate(startDate);
        bal.setEndDate(endDate);
        bal.setStatus(TrialBalanceConstants.CALC_BALANCE_STATUS_UNFIN);
        return bal;
    }

    public TrialBalanceVo periodAmt() {
        try {
            singleTranQueue = new ArrayBlockingQueue<TrialBalanceVo>(queueSize);
            ExecutorService service = Executors.newFixedThreadPool(threadNum);

            String countSql = "select count(DISTINCT GOPAY_ORDER_ID) from ACPS_ACCT_TRANS_DTL_TMP";
            String sql = "SELECT GOPAY_ORDER_ID,1 from ACPS_ACCT_TRANS_DTL_TMP GROUP BY GOPAY_ORDER_ID";
            int total = commonQueryDAO.findCountBySql(countSql);
            logger.info("calculate by gopayOrderId, total {}", total);

            int pageSize = 100;
            int pageNum = (int) Math.ceil((total * 1.0) / pageSize);
            for (int pageIndex = 1; pageIndex <= pageNum; pageIndex++) {
                List<Object[]> orderIdList = commonQueryDAO.findBySql(sql, pageIndex, pageSize);

                for (Object[] orderId : orderIdList) {
                    final String gopayOrderId = (String) orderId[0];
                    pendingVisits.incrementAndGet();
                    service.execute(new Runnable() {
                        public void run() {
                            journalizeAmtTrial(gopayOrderId);
                        }
                    });

                }
            }
            BigDecimal dSum = new BigDecimal(0.0);
            BigDecimal cSum = new BigDecimal(0.0);
            try {
                while (pendingVisits.get() > 0 || singleTranQueue.size() > 0) {
                    if(singleTranQueue.size() > 0) {
                    TrialBalanceVo bd = singleTranQueue.poll(10, TimeUnit.SECONDS);
                    if (bd != null) {
                        if (bd.getDrAmt() != null) {

                            dSum = dSum.add(bd.getDrAmt());
                        } else {
                            logger.error("tranCalculate TrialBalanceVoDrAmt null");
                        }
                        if (bd.getCrAmt() != null) {

                            cSum = cSum.add(bd.getCrAmt());
                        } else {
                            logger.error("tranCalculate TrialBalanceVoCrAmt null");
                        }
                    } else {
                        logger.error("tranCalculate TrialBalanceVo null");
                    }
                }
                }
            } catch (InterruptedException e) {
                logger.error("queue poll error");
                e.printStackTrace();
            } finally {
                service.shutdown();
            }
            TrialBalanceVo vo = new TrialBalanceVo();
            vo.setDrAmt(dSum);
            vo.setCrAmt(cSum);
            return vo;
        } catch (Exception e) {
            logger.error("tranCalculate error");
            StringWriter sw = new StringWriter();
            PrintWriter pw = new PrintWriter(sw);
            e.printStackTrace(pw);
            logger.error(sw.toString());
            return new TrialBalanceVo(new BigDecimal(0.0), new BigDecimal(0.0));
        }
    }

    private void journalizeAmtTrial(final String gopayOrderId) {
        try {

            List<AcpsAcctTransDtlTmp> list = acpsAcctTransDtlTmpDAO.getDtlTmpByGopayOrderId(gopayOrderId);
            for (AcpsAcctTransDtlTmp dtl : list) {
                if (TrialBalanceConstants.DTL_TMP_UNDEAL.equals(dtl.getStatus())) {
                    dtl.setStatus(TrialBalanceConstants.DTL_TMP_DEALING);
                    dtl.setModified(new Date());
                } else {
                    logger.error("journalizeAmtTrial acpsAcctTransDtlTmp {} lock error", dtl.getAcctTransDtlSq());
                    return;
                }
            }
            acpsAcctTransDtlTmpDAO.saveOrUpdateAll(list);

            BigDecimal dAmt = new BigDecimal(0.0);
            BigDecimal cAmt = new BigDecimal(0.0);
            for (AcpsAcctTransDtlTmp dtl : list) {
                if (dtl.getDrCr().equals(Drcr.D)) {
                    dAmt = dAmt.add(dtl.getTranAmt().setScale(2, BigDecimal.ROUND_HALF_UP));
                } else if (dtl.getDrCr().equals(Drcr.C)) {
                    cAmt = cAmt.add(dtl.getTranAmt().setScale(2, BigDecimal.ROUND_HALF_UP));
                }
            }
            TrialBalanceVo vo = new TrialBalanceVo();
            vo.setDrAmt(dAmt);
            vo.setCrAmt(cAmt);
            try {
                singleTranQueue.put(vo);
            } catch (Exception ex) {
                logger.error("singleTranTrial queue put error, gopayOrderId {}", gopayOrderId);
//                ex.printStackTrace();
            }
//            logger.info("queue put success");
            // 记录错误明细
            if (dAmt.compareTo(cAmt) != 0) {
                AcpsCalcBalanceErrDtl dtl = createAcpsCalcBalanceErrDtl(gopayOrderId, dAmt, cAmt);
                acpsCalcBalanceErrDtlDAO.save(dtl);
            }
            // 更新快照状态
            for (AcpsAcctTransDtlTmp dtl : list) {
                dtl.setStatus(TrialBalanceConstants.DTL_TMP_FINISH);
                dtl.setModified(new Date());
            }
            acpsAcctTransDtlTmpDAO.saveOrUpdateAll(list);
        } catch (Exception ex) {
            logger.error("journalizeAmtTrial error");
            logger.error(ex.getMessage());
            StringWriter sw = new StringWriter();
            PrintWriter pw = new PrintWriter(sw);
            ex.printStackTrace(pw);
            logger.error(sw.toString());
            // ex.printStackTrace();
        } finally {
            pendingVisits.decrementAndGet();
        }
    }

    private AcpsCalcBalanceErrDtl createAcpsCalcBalanceErrDtl(final String gopayOrderId, BigDecimal dAmt,
            BigDecimal cAmt) {
        AcpsCalcBalanceErrDtl dtl = new AcpsCalcBalanceErrDtl();
        dtl.setId(acpsCalcBalanceErrDtlDAO.getNextSq());
        dtl.setGopayOrderId(gopayOrderId);
        dtl.setCurrC(cAmt);
        dtl.setCurrD(dAmt);
        dtl.setErrAmt(dAmt.subtract(cAmt).abs());
        dtl.setCalcBalanceId(calcBalanceId);
        dtl.setCreated(new Date());
        dtl.setModified(new Date());
        return dtl;
    }

    public void initializeTmp(Date startDate, Date endDate) {
        acpsAcctTransDtlTmpDAO.deleteByHql("delete from AcpsAcctTransDtlTmp");

        int total = acctTransManager.getAcctTransCountByTimeInterval(startDate, endDate);

        logger.info("initialize AcpsAcctTransDtlTmp, total {}", total);

        int pageSize = 100;
        int pageNum = (int) Math.ceil((total * 1.0) / pageSize);
        for (int pageIndex = 1; pageIndex <= pageNum; pageIndex++) {
            List<AcpsAcctTransDtl> dtlList = acctTransManager.getAcctTransDtlsByTimeInterval(startDate, endDate,
                    pageIndex, pageSize);
            List<AcpsAcctTransDtlTmp> tmpList = new ArrayList<AcpsAcctTransDtlTmp>(100);
            for (AcpsAcctTransDtl dtl : dtlList) {
                AcpsAcctTransDtlTmp tmp = AcpsAcctTransDtl2tmp(dtl);
                tmpList.add(tmp);
            }
            acpsAcctTransDtlTmpDAO.saveOrUpdateAll(tmpList);
        }

    }

    public TrialBalanceVo openBalance(Date startDate, Date endDate) {
        DetachedCriteria queryCriteria = DetachedCriteria.forClass(AcpsCalcBalance.class);
        queryCriteria.add(Restrictions.eq("endDate", DateUtils.add(endDate, Calendar.DATE, -1)));
        // queryCriteria.addOrder(Order.desc("created"));
        List<AcpsCalcBalance> List = acpsCalcBalanceDAO.findByCriteria(queryCriteria, 1, 1);
        TrialBalanceVo vo = new TrialBalanceVo(new BigDecimal(0.0), new BigDecimal(0.0));
        if (List.size() > 0) {
            AcpsCalcBalance bal = List.get(0);
            if (bal != null) {
                vo.setDrAmt(null == bal.getEndDr() ? new BigDecimal(0.0) : bal.getEndDr());
                vo.setCrAmt(null == bal.getEndCr() ? new BigDecimal(0.0) : bal.getEndCr());
            }
        }
        return vo;
    }

    public TrialBalanceVo closingBalance() {
        StringBuilder sql = new StringBuilder();
        sql.append("SELECT to_char(SUM(round((CASE");
        sql.append("             WHEN SCHEMA_CODE = " + drSchema + " THEN");
        // sql.append("             WHEN SCHEMA_CODE = '1021' THEN");
        sql.append("              BAL_AMT");
        sql.append("             ELSE");
        sql.append("              NULL");
        sql.append("           END),2))) D_AMT,");
        sql.append("       to_char(SUM(round((CASE");
        sql.append("             WHEN SCHEMA_CODE IN (" + crSchema + ") THEN");
        // sql.append("             WHEN SCHEMA_CODE IN ('2202', '2011','2241','2002','2801','4103','4104','6021','6011','6411') THEN");
        sql.append("              BAL_AMT");
        sql.append("             ELSE");
        sql.append("              NULL");
        sql.append("           END),2))) C_AMT,");
        sql.append("       SYSDATE");
        sql.append("  FROM ACPS_ACCT_MAST");

        List<Object[]> list = commonQueryDAO.findBySql(sql.toString());
        TrialBalanceVo vo = new TrialBalanceVo(new BigDecimal(0.0), new BigDecimal(0.0));
        if (list.size() > 0) {
            Object[] result = list.get(0);
            if (result != null) {
                vo.setDrAmt(null == result[0] ? new BigDecimal(0.0) : new BigDecimal((String) result[0]));
                vo.setCrAmt(null == result[1] ? new BigDecimal(0.0) : new BigDecimal((String) result[1]));
            }
        }
        // logger.info("day end trial balance...");
        return vo;
    }

    private AcpsAcctTransDtlTmp AcpsAcctTransDtl2tmp(AcpsAcctTransDtl dtl) {
        AcpsAcctTransDtlTmp tmp = new AcpsAcctTransDtlTmp();
        // tmp.setAcctTransDtlSq(Long.valueOf(acpsAcctTransDtlDAO.getNextId(AcpsAcctTransDtl.SEQUENCE_TRAN_SQ)));
        tmp.setAcctTransDtlSq(dtl.getAcctTransDtlSq());
        tmp.setGopayOrderId(dtl.getGopayOrderId());
        tmp.setAcctId(dtl.getAcctId());
        tmp.setCustId(dtl.getCustId());
        tmp.setDrCr(dtl.getDrCr());
        tmp.setTranId(dtl.getTranId());
        tmp.setPartTranSrlNum(dtl.getPartTranSrlNum());
        tmp.setTranTime(dtl.getTranTime());
        tmp.setTranType(dtl.getTranType());
        tmp.setCurType(dtl.getCurType());
        tmp.setTranAmt(dtl.getTranAmt());
        tmp.setCurBalAfterTrans(dtl.getCurBalAfterTrans());
        tmp.setCurBalBeforeTrans(dtl.getCurBalBeforeTrans());
        tmp.setDelFlg(dtl.getDelFlg());
        tmp.setValueDate(dtl.getValueDate());
        tmp.setTranParticular(dtl.getTranParticular());
        tmp.setTranRmks(dtl.getTranRmks());
        tmp.setPstdFlg(dtl.getPstdFlg());
        tmp.setLchgUserId(dtl.getLchgUserId());
        tmp.setLastUpdTm(dtl.getLastUpdTm());
        tmp.setRcreUserId(dtl.getRcreUserId());
        tmp.setRcreTime(dtl.getRcreTime());
        tmp.setDtlPackageId(dtl.getDtlPackageId());
        tmp.setCreated(new Date());
        tmp.setModified(new Date());
        tmp.setStatus(TrialBalanceConstants.DTL_TMP_UNDEAL);
        return tmp;

    }

    public String getDrSchema() {
        return drSchema;
    }

    public void setDrSchema(String drSchema) {
        this.drSchema = drSchema;
    }

    public String getCrSchema() {
        return crSchema;
    }

    public void setCrSchema(String crSchema) {
        this.crSchema = crSchema;
    }

    public int getThreadNum() {
        return threadNum;
    }

    public void setThreadNum(int threadNum) {
        this.threadNum = threadNum;
    }

    public int getQueueSize() {
        return queueSize;
    }

    public void setQueueSize(int queueSize) {
        this.queueSize = queueSize;
    }

}
